package net.wlm.others;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class JourneyRealtimeFunction extends KeyedProcessFunction<String, CarBean, TripDetectionResult> {

    private static final Logger LOG = LoggerFactory.getLogger(JourneyRealtimeFunction.class);

    // 简化的状态定义
    private ValueState<Long> tripStartTimeState;
    private ValueState<Double> startMileageState;

    @Override
    public void open(Configuration conf) {
        tripStartTimeState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("tripStartTime", Long.class));
        startMileageState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("startMileage", Double.class));
    }

    @Override
    public void processElement(CarBean data, Context ctx, Collector<TripDetectionResult> out) throws Exception {
        // 基本数据校验
        if (data.getSpeed() == null || data.getTimestamp() == null || data.getVin() == null) {
            return;
        }

        // 简化逻辑：只要速度>0就认为是行程开始
        if (data.getSpeed() > 0) {
            if (tripStartTimeState.value() == null) {
                tripStartTimeState.update(data.getTimestamp());
                startMileageState.update(data.getMileage());
                LOG.info("Trip started for VIN: {}", data.getVin());
            }
        }
        // 简化逻辑：只要速度=0就认为是行程结束
        else {
            if (tripStartTimeState.value() != null) {
                TripDetectionResult result = new TripDetectionResult();
                // 设置结果中的车架号
                result.setVin(data.getVin());
                // 获取行程ID
                result.setTripId(data.getVin() + "-" + tripStartTimeState.value());
                // 获取结果开始时间
                result.setStartTime(tripStartTimeState.value());
                // 设置结果的结束时间
                result.setEndTime(data.getTimestamp());
                // 计算并设置行驶距离，通过当前里程数减去起始里程数得到
                result.setDistance(data.getMileage() - startMileageState.value());
                out.collect(result);
                LOG.info("Trip ended for VIN: {}, Distance: {} km", data.getVin(), result.getDistance());
                // 清除状态
                tripStartTimeState.clear();
                startMileageState.clear();
            }
        }
    }
}